package edu.vt.ece.rt.quorum;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import org.jgroups.Address;
import org.jgroups.Global;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.jgroups.conf.ConfiguratorFactory;
import org.jgroups.conf.ProtocolStackConfigurator;
import org.jgroups.protocols.BARRIER;
import org.jgroups.protocols.FD;
import org.jgroups.protocols.FD_ALL;
import org.jgroups.protocols.FD_SOCK;
import org.jgroups.protocols.FRAG2;
import org.jgroups.protocols.MERGE2;
import org.jgroups.protocols.MFC;
import org.jgroups.protocols.PING;
import org.jgroups.protocols.TP;
import org.jgroups.protocols.UDP;
import org.jgroups.protocols.UFC;
import org.jgroups.protocols.UNICAST2;
import org.jgroups.protocols.VERIFY_SUSPECT;
import org.jgroups.protocols.pbcast.GMS;
import org.jgroups.protocols.pbcast.NAKACK;
import org.jgroups.protocols.pbcast.STABLE;
import org.jgroups.protocols.pbcast.STATE_TRANSFER;
import org.jgroups.stack.Protocol;
import org.jgroups.stack.ProtocolStack;

import edu.vt.ece.rt.quorum.messsages.MessageHandler;
import edu.vt.ece.rt.quorum.tree.Node;
import edu.vt.ece.rt.quorum.tree.Tree;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.util.Logger.Logger;

/*
 * Basic node class allowing the node to communicate with other node
 * 
 * @author Sudhanshu 
 */

public class Cluster extends ReceiverAdapter{
	
	/*
	 * base Channel allows to create a cluster of all the nodes, helpful in detecting
	 * any node failure across the cluster.
	 */
	JChannel channel;
	
	/*
	 * Local node address 
	 */
	Address myAddress;
	
	/*
	 * All the members of cluster
	 */
	public View currentView;
	
	/*
	 * Name of multicast group
	 */
	String group;
	
	/*
	 * Tree structure to create quorums from
	 */
	Tree tree;
	
	/*
	 * Node List, to access the quorums as List
	 */
	List<Node> nodeList;
	
	public Cluster(){
		group = Names.base;
	}
	
	public Cluster(String s){
		group = s;
	}
	
	/*
	 * To initiate a node and make it part of cluster "Quorum"
	 */
	public void init() throws Exception{
//		channel = new JChannel("tcp.xml");  		//Create a basic UDP Protocol Stack based Channel
		
		ProtocolStackConfigurator configs = ConfiguratorFactory.getStackConfigurator("udp.xml");
		
		configs.getProtocolStack().get(0).getProperties().put(Global.SINGLETON_NAME, "instance");
		
		
		
		channel = new JChannel(configs);
		NAKACK nck = (NAKACK)channel.getProtocolStack().findProtocol("NAKACK");
		if (nck != null) {
			System.out.println("Hey NAKACK2 is there");
		} else {
			System.out.println("Hey NAKACK2 is not there XXXXXXXXX");
		}

//		ProtocolStack stack=channel.getProtocolStack(); // 2
//		UDP udp = (UDP)stack.findProtocol("UDP");
//		udp.setValue("singleton_name", "single");
		              // 3
		
//		JChannel dummy = new JChannel();
//		ProtocolStack stack=dummy.getProtocolStack(); // 2
//		UDP udp = (UDP)stack.findProtocol("UDP");
//		
////		udp.setValue("ip_ttl", 8);
//		udp.setValue("singleton_name", "single");
//		stack.init();
//		stack.initProtocolStack();
//		channel.setProtocolStack(stack);
//		stack.addProtocol(udp)
//		     .addProtocol(new PING())
//		     .addProtocol(new MERGE2())
//		     .addProtocol(new FD_SOCK())
//		     .addProtocol(new FD_ALL().setValue("timeout", 12000))
//		     .addProtocol(new VERIFY_SUSPECT())
//		     .addProtocol(new BARRIER())
//		     .addProtocol(new NAKACK())
//		     .addProtocol(new UNICAST2())
//		     .addProtocol(new STABLE())
//		     .addProtocol(new GMS())
//		     .addProtocol(new UFC())
//		     .addProtocol(new MFC())
//		     .addProtocol(new FRAG2())
//		     .addProtocol(new STATE_TRANSFER());       // 4
		
		for (String keys : (ProtocolStack.getSingletonTransports()).keySet()){
			System.out.println("Singleton "+keys);
		}
		
//		stack.init();        
		
		
		
		
		
		ProtocolStack stk = channel.getProtocolStack();
//		TP t = stk.getTransport();
		
//		if (ClustersManager.tp == null) {
//			ClustersManager.tp = t;
//			System.out.println("Setting for first time");
//		} else if (ClustersManager.tp == t) {
//			System.out.println("Wow it is the same");
//		} else {			
//			t.setValue("singleton_name", "single");
//			System.out.println("singleton name is "+ t.getValue("singleton_name"));	
			
//		}
		
//		s.get(0).getProperties().put(Global.SINGLETON_NAME,
//				"instance");
//		ProtocolStack.getSingletonTransports()).get
		
//		System.out.println("Tried to print singleton values");
//		stk.getTransport().setValue("singleton_name", "single");
		
		Protocol p = stk.findProtocol("UDP");
		
		
		FD_ALL p1 = (FD_ALL) stk.findProtocol("FD_ALL");
		FD p2 = (FD) stk.findProtocol("FD");
		if (p2 != null)p2.stopFailureDetection();

		if (p1 != null) {
//			System.out.println("Found FD_ALL");
			
//			p1.setTimeout(100000);
//			p1.setInterval(100000);
//			p1.setValue("timeout", 100000);
//			p1.setValue("interval", 100000);
////			p1.setValue("max_tries", 100);
////			p1.setValue("shun", false);
//			p1.stopFailureDetection();
		} else {
			System.out.println("Could not find FD_ALL");
		}
		if(p!=null){
			p.setLevel(System.getProperty(Names.jLogger));
			p.setValue("ip_ttl", 0);
			System.out.println("TTL value is "+p.getValue("ip_ttl"));
			
		}
		
//		Logger.verbose = Integer.parseInt(System.getProperty(Names.verbose));		
		channel.setReceiver(this);
        channel.connect(group);			//Add to multicast group Quorum
        System.out.println("Current member size for "+ group +" is "+ channel.getView().size());
        if (channel.getView().size() == 1) {
        	System.out.println(channel.getAddress()+" is the coordinator");
        } else {
        	System.out.println(channel.getAddress()+" is not the first one");
        }
        channel.setDiscardOwnMessages(true);		//By Default no self receive of messages
        channel.getState(null, 10000);
        myAddress = channel.getAddress();   
        Logger.debug("My address "+myAddress);
	}
	

	/*
	 * To wait on other nodes in network to get initiated
	 */
	public void waitForCluster(){
		waitForCluster(Integer.getInteger(Names.nodes));
	}
	public void waitForCluster(Integer nodes) {
		Boolean t = true;
//		if (ClustersManager.baseClusterId == 0){
//			while(t);
//		}
	     while(currentView.getMembers().size() != nodes) {
	    	 try {
				Thread.sleep(20000L);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	     }
         
//         if(group.equals(Names.base))
//                 setNodeId();
    }

	private void setNodeId() {
		//Set the Node ID
		int i=0;
		for(Address adr: currentView.getMembers()){
			if(adr.equals(myAddress))
				break;
			i++;
		}
		ClustersManager.baseClusterId = i;
		Logger.debug("My ID "+i);
	}

	/*
	 *  To Close channel gracefully
	 */
	public void close(){
		channel.close();
	}

	@Override
	public void viewAccepted(View paramView) {
		currentView = paramView;		
		/*
		 * Check for reductions in members, if yes, trigger re-run of quorum protocol
		 * TODO: Other way, use the mute message to stop the node to talk other
		 */
	}
	
	public void createTree(int degree) {
		nodeList = new ArrayList<Node>();	
		for(Address adr: currentView.getMembers()){
			nodeList.add(new Node(adr));
		}
		ArrayList<Node> tmp = new ArrayList<Node>();
		for(Node n: nodeList)
			tmp.add(n);
		
		tree = Tree.createTree(tmp, degree);
	} 
	
	/*
	 * Depending on message it is send to group or unicast to particular member
	 */
	public void send(Message msg) throws Exception{
		channel.send(msg);
	}
	
	/*
	 * (non-Javadoc)
	 * @see org.jgroups.ReceiverAdapter#receive(org.jgroups.Message)
	 */
    public void receive(Message msg) {
    	MessageHandler.process(msg);
    }

    public int getReplyCount(){
    	//NOTE: Might me incorrect if implementation differ
    	return currentView.getMembers().size();
    }
    
    public Tree getTree(){
    	if(tree == null){
    		createTree(Integer.getInteger(Names.degree));
    	}
    	return tree;
    }
    
    public Address getMyAddress(){
    	return myAddress;
    }
    
    public void setAutoReceive(boolean b)	{
    	channel.setDiscardOwnMessages(!b);
    }
    
    public boolean getAutoReceive(){
    	return channel.getDiscardOwnMessages();
    }
    
    /*
     * Test the Cluster class, TODO: crude way, create JUnit tests
     */
    public static void main(String[] args) throws Exception {
		Cluster c = new Cluster();
		
		c.init();
		
		Logger.debug("Sending Message");
		c.setAutoReceive(true);
		c.send(new Message(null,"Hello there"));
		
		c.close();
	}
}
